package com.kancy.streamplus.service.impl;

import com.kancy.streamplus.channel.DynamicChannelResolver;
import com.kancy.streamplus.channel.DynamicDelayPollingMessage;
import com.kancy.streamplus.channel.PollingChannel;
import com.kancy.streamplus.channel.PollingMessage;
import com.kancy.streamplus.properties.PollingMessageProperties;
import com.kancy.streamplus.properties.StreamPlusProperties;
import com.kancy.streamplus.service.PollingMessageService;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * PollingMessageServiceImpl
 *
 * @author kancy
 * @date 2020/3/22 15:02
 */
public class PollingMessageServiceImpl implements PollingMessageService {

    private final StreamPlusProperties streamPlusProperties;
    private final DynamicChannelResolver dynamicChannelResolver;
    private final PollingChannel pollingChannel;

    public PollingMessageServiceImpl(StreamPlusProperties streamPlusProperties,
                                     DynamicChannelResolver dynamicChannelResolver,
                                     PollingChannel pollingChannel) {
        this.streamPlusProperties = streamPlusProperties;
        this.dynamicChannelResolver = dynamicChannelResolver;
        this.pollingChannel = pollingChannel;
    }

    /**
     * 发送消息
     *
     * @param pollingMessage
     */
    @Override
    public boolean sendMessage(PollingMessage pollingMessage) {
        Assert.notNull(pollingMessage, "message payload not allowed to be null");
        Assert.isTrue(!StringUtils.isEmpty(pollingMessage.getMessageKey()),
                String.format("PollingMessage [%s] message key is empty", pollingMessage.getClass().getName()));

        PollingMessageProperties properties = getPollingMessageProperties(pollingMessage);
        Map<String, Object> headers = initMessageHeader(pollingMessage, properties);
        long delay = properties.getDelay().toMillis();
        if (pollingMessage instanceof DynamicDelayPollingMessage){
            delay = DynamicDelayPollingMessage.class.cast(pollingMessage).getDelay().toMillis();
        }
        headers.put(PollingMessage.DELAY, delay);
        if (delay > 0){
            // 如果是延迟消息，添加0.5秒程序执行或者网络传输过程的误差
            headers.put(PollingMessage.LAST_QUERY_TIMESTAMP, System.currentTimeMillis() + 500);
            return pollingChannel.output().send(MessageBuilder.withPayload(pollingMessage).copyHeaders(headers).build());
        }
        return dynamicChannelResolver.getChannel(properties.getChannelName()).send(
                MessageBuilder.withPayload(pollingMessage).copyHeaders(headers).build());
    }

    private Map<String, Object> initMessageHeader(PollingMessage pollingMessage, PollingMessageProperties properties) {
        Map<String, Object> headers = new HashMap<>();
        headers.put(PollingMessage.MESSAGE_KEY, pollingMessage.getMessageKey());
        headers.put(PollingMessage.CHANNEL_NAME, properties.getChannelName());
        headers.put(PollingMessage.MESSAGE_TYPE, properties.getMessageType());
        headers.put(PollingMessage.MESSAGE_HANDLER, properties.getMessageHandler());
        headers.put(PollingMessage.FREQUENCY, properties.getFrequency());
        headers.put(PollingMessage.MAX_LIVE_TIME, properties.getMaxLiveTime());
        headers.put(PollingMessage.MAX_RETRY_TIMES, properties.getMaxRetryTimes());
        headers.put(PollingMessage.RETRY_COUNT, 0);
        headers.put(PollingMessage.FIRST_QUERY_TIMESTAMP, System.currentTimeMillis());
        headers.put(PollingMessage.LAST_QUERY_TIMESTAMP, System.currentTimeMillis());
        return headers;
    }

    private PollingMessageProperties getPollingMessageProperties(PollingMessage pollingMessage) {
        return streamPlusProperties.getPollingMessage().get(pollingMessage.getMessageKey());
    }

}
